package com.shujia.flink.window;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.annotation.JSONField;
import lombok.AllArgsConstructor;
import lombok.Data;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 实时统计卡口拥堵情况
 * 1、统计最近15分钟数据，每隔1分钟统计一次
 * 2、统计卡口平均车速和车流量
 *
 * @author shujia
 */
public class Demo10CardFlow {
    public static void main(String[] args) throws Exception {

        //创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //读取数据
        DataStream<String> lines = env.socketTextStream("master", 8888);

        //解析数据
        DataStream<Car> cars = lines.map(line -> {
            //将json字符串转换成对象
            return JSON.parseObject(line, Car.class);
        }, TypeInformation.of(Car.class));

        //指定事件时间和水位线
        WatermarkStrategy<Car> watermarkStrategy = WatermarkStrategy
                //指定水位线前移时间
                .<Car>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                //指定时间字段
                .withTimestampAssigner((car, ts) -> car.getTime() * 1000);

        DataStream<Car> carsAss = cars.assignTimestampsAndWatermarks(watermarkStrategy);

        ///按照卡口分组
        KeyedStream<Car, Long> keyBy = carsAss.keyBy(Car::getCard);

        //划分窗口
        WindowedStream<Car, Long, TimeWindow> carsWindow = keyBy.window(SlidingEventTimeWindows.of(Time.minutes(60), Time.minutes(1)));

        //统计卡口平均车速和车流量
        DataStream<Result> results = carsWindow
                .process(new ProcessWindowFunction<Car, Result, Long, TimeWindow>() {
                    /**
                     * process每一个key每个窗口执行一次
                     * @param card 卡口编号
                     * @param context 上下文对象,可以获取窗口时间,水位线
                     * @param elements 一个窗口内所有的数据
                     * @param out 用于将数据发生到下游
                     */
                    @Override
                    public void process(Long card,
                                        ProcessWindowFunction<Car, Result, Long, TimeWindow>.Context context,
                                        Iterable<Car> elements,
                                        Collector<Result> out) {
                        long flow = 0;
                        double sumSpeed = 0;
                        for (Car element : elements) {
                            //计算总的车速
                            sumSpeed = sumSpeed + element.getSpeed();
                            //计算车流量
                            flow++;
                        }

                        //计算平均车速
                        double avgSpeed = sumSpeed / flow;

                        //获取窗口的开始和结束时间
                        long start = context.window().getStart();
                        long end = context.window().getEnd();

                        //将结果发生到下游
                        out.collect(new Result(card, flow, sumSpeed, avgSpeed, start, end));
                    }
                });

        results.print();

        env.execute();

    }

    @Data
    @AllArgsConstructor
    static class Car {
        //{"car":"皖A9A7N2",
        // "city_code":"340500",
        // "county_code":"340522",
        // "card":117988031603010,
        // "camera_id":"01001",
        // "orientation":"西南",
        // "road_id":34052056,
        // "time":1614711904,
        // "speed":35.38}
        private String car;
        @JSONField(name = "city_code")
        private String cityCode;

        @JSONField(name = "county_code")
        protected String countyCode;
        private long card;
        @JSONField(name = "camera_id")
        private String cameraDd;
        private String orientation;

        @JSONField(name = "road_id")
        private long roadDd;
        private long time;
        private double speed;
    }

    @Data
    @AllArgsConstructor
    static class Result {
        Long card;
        long flow;
        double sumSpeed;
        double avgSpeed;
        long start;
        long end;
    }
}
